package top.lywivan.netty.reactor.cnxn;


import top.lywivan.netty.reactor.Server;
import top.lywivan.netty.reactor.serialize.JdkSerializer;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;

public class NIOServerCnxnFactory extends ServerCnxnFactory{

    public static final String NIO_NUM_SELECTOR_THREADS = "nio.numSelectorThreads";
    /** Default: 2 * numCores */
    public static final String NIO_NUM_WORKER_THREADS = "nio.numWorkerThreads";


    // 客户端最大连接数限制
    protected int maxClientCnxns = 60;

    int listenBacklog = -1;

    private int numSelectorThreads;
    private int numWorkerThreads;//IO工作线程池中的线程数


    private AcceptThread acceptThread;
    private final Set<SelectorThread> selectorThreads = new HashSet<SelectorThread>();
    /**
     * 处理IO操作的工作线程池；线程池数由numWorkerThreads决定
     */
    protected IOWorkerService workerPool;


    ServerSocketChannel ss;


    protected Server server;

    // 是否要停止服务的标识
    private volatile boolean stopped = true;

    @Override
    public void configure(InetSocketAddress clientPortAddress, int maxClientCnxns, int clientPortListenBacklog, boolean secure) throws IOException{
        this.maxClientCnxns = maxClientCnxns;
        initMaxCnxns();

        //准备相关资源
        int coreNum = Runtime.getRuntime().availableProcessors();
        // 计算用于检测客户端IO事件的线程数量 2
        numSelectorThreads = Integer.getInteger(NIO_NUM_SELECTOR_THREADS,
                Math.max((int) Math.sqrt((double) coreNum / 2 ),1));
        // 计算用于处理客户端IO事件的线程数量
        numWorkerThreads = Integer.getInteger(NIO_NUM_WORKER_THREADS,2 * coreNum);


        //准备好SelectThread
        for (int i=0;i<numSelectorThreads;i++) {
            selectorThreads.add(new SelectorThread(i));
        }

        listenBacklog = clientPortListenBacklog;

        // 打开一个服务端的ServerSocketChannel
        this.ss = ServerSocketChannel.open();
        this.ss.configureBlocking(false);
        ss.socket().setReuseAddress(true);
        if (listenBacklog == -1) {
            ss.socket().bind(clientPortAddress);
        } else {
            ss.socket().bind(clientPortAddress, listenBacklog);
        }
        //创建AcceptThrad
        acceptThread = new AcceptThread(this.ss,selectorThreads,clientPortAddress);
    }

    @Override
    public void startup(Server server, boolean startServer) throws IOException, InterruptedException {
        //启动Reactor接收客户端的入口
        start();
        setServer(server);
        if (startServer) {
            // TODO load data....

            //start other component
            server.startup();
        }
    }

    private void setServer(Server server) {
        this.server = server;
        if (server!=null) {
            server.setServerCnxnFactory(this);
            // 设置序列化
            server.setSerializer(new JdkSerializer());
        }

    }

    private void start() {
        stopped = false;

        //启动IO工作线程池
        if (workerPool == null) {
            workerPool = new IOWorkerService(numWorkerThreads,"IOWorker");
        }

        //启动selector thread
        for (SelectorThread selectorThread : selectorThreads) {
            if (selectorThread.getState() == Thread.State.NEW) {
                selectorThread.start();
            }
        }
        // 启动accept thread
        if (acceptThread.getState() == Thread.State.NEW) {
            acceptThread.start();
        }
    }


    ////////////////////////////////Thread
    private abstract class AbstractSelectThread extends ServerThread {

        protected final Selector selector;

        public AbstractSelectThread(String threadName) throws IOException {
            super(threadName);
            //后台线程
            setDaemon(true);
            // 打开当前线程的Selector
            selector = Selector.open();
        }


        public void wakeupSelector() {
            selector.wakeup();
        }

        protected void cleanupSelectionKey(SelectionKey key) {
            if (key != null) {
                try {
                    key.cancel();
                } catch (Exception ex) {
                }
            }
        }

        protected void fastCloseSock(SocketChannel sc) {
            if (sc != null) {
                try {
                    // Hard close immediately, discarding buffers
                    sc.socket().setSoLinger(true, 0);
                } catch (SocketException e) {
                }
                NIOServerCnxn.closeSock(sc);
            }
        }


    }
    // 接收客户端连接
    private class AcceptThread extends AbstractSelectThread {

        protected void closeSelector() {
            try {
                selector.close();
            } catch (IOException e) {
            }
        }

        private final ServerSocketChannel serverSocketChannel;
        private final SelectionKey acceptKey;
        private final Collection<SelectorThread> selectorThreads;// 2
        private Iterator<SelectorThread> selectorIterator;

        public AcceptThread(ServerSocketChannel ss, Set<SelectorThread> selectorThreads, InetSocketAddress address) throws IOException {
            super("NIOServerCxnFactory.AcceptThread:" + address);
            this.serverSocketChannel = ss;
            this.acceptKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            this.selectorThreads = Collections.unmodifiableList(new ArrayList<>(selectorThreads));
            this.selectorIterator = selectorThreads.iterator();
        }

        /**
         * AcceptThread 检测是否有新连接
         */
        @Override
        public void run() {
            try {
                while (!stopped && !serverSocketChannel.socket().isClosed()) {
                    //执行select程序
                    select();
                }
            } finally {
                closeSelector();
            }
        }

        private void select() {
            try {
                selector.select();
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (!stopped && iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();

                    if (!key.isValid()) {
                        continue;
                    }
                    // 判断是否是accept
                    if (key.isAcceptable()) {
                        // 接收连接
                        if (!doAccept()) {
                            pauseAccept(10);
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        /**
         * 接收新连接,并绑定到一个selector thread
         * @return
         */
        private boolean doAccept() {
            boolean accepted = false;
            SocketChannel socketChannel = null;
            try {
                socketChannel = serverSocketChannel.accept();
                accepted = true;

                // TODO  cnxn over limit

                // 设置非阻塞
                socketChannel.configureBlocking(false);
                // 将当前连接 绑定给一个 selector thread round robin
                if (!selectorIterator.hasNext()) {
                    selectorIterator = selectorThreads.iterator();
                }
                SelectorThread selectorThread = selectorIterator.next();
                if (!selectorThread.addAcceptedConnection(socketChannel)) {
                    throw new IOException("unable to add connection to selector thread");
                }
            } catch (IOException e) {
                e.printStackTrace();
                fastCloseSock(socketChannel);
            }
            return accepted;
        }


        private void pauseAccept(long millisecs) {
            acceptKey.interestOps(0);
            try {
                selector.select(millisecs);
            } catch (IOException e) {
                // ignore
            } finally {
                acceptKey.interestOps(SelectionKey.OP_ACCEPT);
            }
        }
    }

    // 检测客户端连接的IO事件
    public class SelectorThread extends AbstractSelectThread {

        private final int id;

        private final Queue<SocketChannel> acceptedQueue;

        private final Queue<SelectionKey> updateQueue;

        public SelectorThread(int id) throws IOException {
            super("NIOServerCxnFactory.SelectorThread-" + id);
            this.id = id;
            acceptedQueue = new LinkedBlockingQueue<>();
            updateQueue = new LinkedBlockingQueue<SelectionKey>();
        }

        // 将新接收的连接存入 acceptedQueue
        public boolean addAcceptedConnection(SocketChannel socketChannel) {
            if (!stopped && !acceptedQueue.offer(socketChannel)) {
                return false;
            }
            // 如果有新连接进来,尝试唤醒selector
            wakeupSelector();
            return true;
        }

        // 将 key 放入队列,以便重新设置感兴趣的事件
        public boolean addInterestOpsUpdateRequest(SelectionKey sk) {
            if (stopped || !updateQueue.offer(sk)) {
                return false;
            }
            wakeupSelector();
            return true;
        }

        @Override
        public void run() {
            while (!stopped) {
                try {
                    // IO select
                    select();
                    // 处理新接收的连接
                    processAcceptedConnections();

                    processInterestOpsUpdateRequests();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            //
            // Close connections still pending on the selector. Any others
            // with in-flight work, let drain out of the work queue.

            for (SelectionKey key : selector.keys()) {
                NIOServerCnxn cnxn = (NIOServerCnxn) key.attachment();
                if (cnxn.isSelectable()) {
                    cnxn.close(ServerCnxn.DisconnectReason.SERVER_SHUTDOWN);
                }
                cleanupSelectionKey(key);
            }
            SocketChannel accepted;
            while ((accepted = acceptedQueue.poll()) != null) {
                fastCloseSock(accepted);
            }
        }

        private void select() {
            try {
                selector.select();
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (!stopped && iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();

                    if (!key.isValid()) {
                        cleanupSelectionKey(key);
                        continue;
                    }

                    //检测是否有 IO
                    if (key.isReadable() || key.isWritable()) {
                        handleIO(key);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        // 包装 selectionKey, 提交给IOWorkPool
        private void handleIO(SelectionKey key) {
            // 封装 IOWorkRequest
            IOWorkRequest ioWorkRequest = new IOWorkRequest(this,key);

            NIOServerCnxn cnxn = (NIOServerCnxn) key.attachment();

            // *** Stop selecting this key while processing on its connection
            cnxn.disableSelectable();
            // IO处理是在异步线程中,Select线程会不断select该key ,在当前IO未完成前暂停检测
            key.interestOps(0);

            //交给 workerPool
            workerPool.schedule(ioWorkRequest);
        }

        private void processAcceptedConnections() {
            SocketChannel sc ;
            while (!stopped && (sc = acceptedQueue.poll()) !=null) {
                SelectionKey key = null;
                try {
                    key = sc.register(selector, SelectionKey.OP_READ);
                    // 封装连接
                    NIOServerCnxn cnxn = createConnection(sc,key,this);
                    key.attach(cnxn);

                } catch (IOException e) {
                    e.printStackTrace();

                    cleanupSelectionKey(key);
                    fastCloseSock(sc);
                }
            }
        }

        // updateQueue 中的 key 重新设置 感兴趣的事件
        private void processInterestOpsUpdateRequests() {
            SelectionKey key;
            while (!stopped && (key = updateQueue.poll()) != null) {
                if (!key.isValid()) {
                    cleanupSelectionKey(key);
                }
                NIOServerCnxn cnxn = (NIOServerCnxn) key.attachment();
                if (cnxn.isSelectable()) {
                    key.interestOps(cnxn.getInterestOps());
                }
            }
        }

    }


    private NIOServerCnxn createConnection(SocketChannel sc, SelectionKey key, SelectorThread selectorThread) throws IOException {
        return new NIOServerCnxn(server,sc,key,this,selectorThread);
    }


    private class IOWorkRequest extends IOWorkerService.IOWork {

        private final SelectorThread selectorThread;
        private final SelectionKey key;
        private final NIOServerCnxn cnxn;

        IOWorkRequest(SelectorThread selectorThread, SelectionKey key) {
            this.selectorThread = selectorThread;
            this.key = key;
            this.cnxn = (NIOServerCnxn)key.attachment();
        }

        // 真正处理 IO
        @Override
        public void doWork() throws Exception {
            if (!key.isValid()) {
                selectorThread.cleanupSelectionKey(key);
                return;
            }

            if (key.isReadable() || key.isWritable()) {
                // 执行IO操作
                cnxn.doIO(key);

                if (stopped) {
                    cnxn.close(ServerCnxn.DisconnectReason.SERVER_SHUTDOWN);
                    return;
                }
                if (!key.isValid()) {
                    selectorThread.cleanupSelectionKey(key);
                    return;
                }
            }

            // Mark this connection as once again ready for selection
            cnxn.enableSelectable();
            cnxn.requestInterestOpsUpdate();
        }


        public void cleanup() {
            cnxn.close(ServerCnxn.DisconnectReason.CLEAN_UP);
        }
    }


}
